package net.jueb.frameworkDemo.se.observer.eventBus2;

import java.util.Iterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 事件总线
 * @author juebanlin
 */
public class EventBusImpl implements EventBus{
	private static final Logger logger = LoggerFactory.getLogger(EventBusImpl.class);
	private ConcurrentHashMap<String, CopyOnWriteArraySet<Receiver<?>>> receivers = new ConcurrentHashMap<>();
	private ScheduledExecutorService scheduler=new ScheduledThreadPoolExecutor(2);
	private Integer queueSize = Integer.valueOf(10000);
	private BlockingQueue<Event<?>> eventQueue;
	private Integer poolSize = Integer.valueOf(1);
	private Integer poolMaxSize = Integer.valueOf(10);
	private Integer poolKeepAlive = Integer.valueOf(60);
	private ExecutorService pool;
	private Runnable consumerRunner = new Runnable() {
		public void run() {
			while (true) {
				while (true) {
					try {
						Event<?> e = EventBusImpl.this.eventQueue.take();
						String name = e.getName();
						if (!EventBusImpl.this.receivers.containsKey(name)) {
							EventBusImpl.logger.warn("事件\'{}\'没有对应的接收器", name);
						} else {
							Iterator<Receiver<?>> it = (EventBusImpl.this.receivers.get(name)).iterator();
							while (it.hasNext()) {
								Receiver<?> receiver =it.next();
								Runnable runner = this.createRunner(receiver, e);

								try {
									EventBusImpl.this.pool.submit(runner);
								} catch (RejectedExecutionException arg6) {
									EventBusImpl.logger.error("事件线程池已满，请尽快调整配置参数");
									this.onRejected(receiver, e);
								}
							}
						}
					} catch (InterruptedException arg7) {
						EventBusImpl.logger.error("获取事件对象时出现异常", arg7);
					}
				}
			}
		}

		@SuppressWarnings("unchecked")
		private void onRejected(Receiver<?> receiver,@SuppressWarnings("rawtypes") Event event) {
			try {
				receiver.onEvent(event);
			} catch (ClassCastException arg3) {
				EventBusImpl.logger.error("事件[{}]对象类型不符合接收器声明,{}", event.getName(), arg3);
			} catch (Throwable arg4) {
				EventBusImpl.logger.error("执行处理事件时发生异常", arg4);
			}

		}
		@SuppressWarnings("unchecked")
		private Runnable createRunner(@SuppressWarnings("rawtypes") final Receiver receiver, @SuppressWarnings("rawtypes") final Event event) {
			return new Runnable() {
				public void run() {
					try {
						receiver.onEvent(event);
					} catch (ClassCastException arg1) {
						EventBusImpl.logger.error("事件[{}]对象类型不符合接收器声明,{}", event.getName(), arg1);
					} catch (Throwable arg2) {
						EventBusImpl.logger.error("事件\'" + event.getName() + "\'处理时发生异常", arg2);
					}

				}
			};
		}
	};
	private volatile boolean stop;

	public EventBusImpl() {
		initialize();
	}
	protected void initialize() {
		NamedThreadFactory threadFactory = new NamedThreadFactory(new ThreadGroup("事件模块"), "事件处理");
		this.eventQueue = new LinkedBlockingQueue<>(this.queueSize.intValue());
		this.pool = new ThreadPoolExecutor(this.poolSize.intValue(), this.poolMaxSize.intValue(),
				(long) this.poolKeepAlive.intValue(), TimeUnit.SECONDS,
				new LinkedBlockingQueue<Runnable>(this.queueSize.intValue()), threadFactory);
		Thread consumer = new Thread(this.consumerRunner, "消费事件后台线程");
		consumer.setDaemon(true);
		consumer.start();
	}

	public void shutdown() {
		if (!this.isStop()) {
			this.stop = true;

			while (!this.eventQueue.isEmpty()) {
				Thread.yield();
			}

			this.pool.shutdown();

			while (!this.pool.isTerminated()) {
				Thread.yield();
			}

		}
	}

	public boolean isStop() {
		return this.stop;
	}

	public void post(Event<?> event) {
		if (event == null) {
			throw new IllegalArgumentException("事件对象不能为空");
		} else if (this.isStop()) {
			throw new IllegalStateException("事件总线已经停止，不能再接收事件");
		} else {
			try {
				this.eventQueue.put(event);
			} catch (InterruptedException arg2) {
				logger.error("在添加事件对象时产生异常", arg2);
			}

		}
	}

	public ScheduledFuture<?> post(Event<?> event, long delay) {
		if (event == null) {
			logger.error("事件对象不能为空");
			return null;
		} else {
			if (this.isStop()) {
				logger.error("事件总线已经停止，不能再接收事件[{}]", event.getName());
			}

			logger.debug("提交延迟[{}毫秒]的定时事件[{}]", event.getName());
			return scheduler.scheduleWithFixedDelay(getScheduleTask(event),0, delay, TimeUnit.MILLISECONDS);
		}
	}

	private Runnable getScheduleTask(final Event<?> event) {
		return new Runnable() {
			public void run() {
				EventBusImpl.this.post(event);
			}
		};
	}

	public void register(String name, Receiver<?> receiver) {
		if (name != null && receiver != null) {
			CopyOnWriteArraySet<Receiver<?>> set = this.receivers.get(name);
			if (set == null) {
				set = new CopyOnWriteArraySet<Receiver<?>>();
				CopyOnWriteArraySet<Receiver<?>> prev = this.receivers.putIfAbsent(name, set);
				set = prev != null ? prev : set;
			}
			set.add(receiver);
		} else {
			throw new IllegalArgumentException("事件名和接收者均不能为空");
		}
	}

	public void unregister(String name, Receiver<?> receiver) {
		if (name != null && receiver != null) {
			CopyOnWriteArraySet<Receiver<?>> set =this.receivers.get(name);
			if (set != null) {
				set.remove(receiver);
			}
		} else {
			throw new IllegalArgumentException("事件名和接收者均不能为空");
		}
	}
}